#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt="{{ execution_date | cst_ds }}"
name='派件员集中签收情况分析'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "日期\t代理区\t网点名称\t收派员\t签收量\t3分钟内集中签收次数">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 10 --executor-memory 12g --executor-cores 4  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
WITH waybill_info AS(
    SELECT
        waybill_no,--运单号
        receiver_mobile_phone --收件人手机号
    FROM jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
    WHERE dt BETWEEN DATE_SUB('{{ execution_date | cst_ds }}',30) AND '{{ execution_date | cst_ds }}'
),
deliver_info AS(
SELECT
    sign.dt, --日期
    sign.waybill_no,--运单号
    sign.site_agent_code,--代理区code
    sign.site_agent_name,--代理区
    sign.signed_site_code,--扫描网点code
    sign.signed_site,--扫描网点
    sign.send_user_code,--派件员编号
    sign.deliver_user,--派件员
    sign.signed_time, --签收时间
    info.receiver_mobile_phone--收件人手机号
FROM jms_dwd.dwd_tab_barscan_sign_base_dt sign
LEFT JOIN waybill_info info
    ON sign.waybill_no = info.waybill_no
WHERE sign.dt = '{{ execution_date | cst_ds }}'
),
abnormal_waybill AS(
    SELECT
        waybill_no
    FROM jms_dwd.dwd_wide_abnormal_detail_waybill_dt
    WHERE dt BETWEEN DATE_SUB('{{ execution_date | cst_ds }}',30) AND '{{ execution_date | cst_ds }}'
    AND is_reback=1
),
dense_deliver_info AS(
    SELECT
        deliver_info.dt, --日期
        abnormal_waybill.waybill_no,--运单号
        deliver_info.site_agent_code,--代理区code
        deliver_info.site_agent_name,--代理区
        deliver_info.signed_site_code,--扫描网点code
        deliver_info.signed_site,--扫描网点
        deliver_info.send_user_code,--派件员编号
        deliver_info.deliver_user,--派件员
        deliver_info.signed_time, --签收时间
        deliver_info.receiver_mobile_phone--收件人手机号
    FROM deliver_info
    LEFT JOIN abnormal_waybill ON deliver_info.waybill_no=abnormal_waybill.waybill_no
),
res_pre AS(
    SELECT
        dt,
        site_agent_code,
        site_agent_name,
        signed_site_code,
        signed_site,
        send_user_code,
        deliver_user,
        COUNT(1) total_cnt_per_three_minutes,
        COUNT(DISTINCT IF(waybill_no IS NULL,receiver_mobile_phone,NULL)) deliver_cnt_per_three_minute
    FROM dense_deliver_info
    GROUP BY
        dt,
        site_agent_code,
        site_agent_name,
        signed_site_code,
        signed_site,
        send_user_code,
        deliver_user,
        CAST(signed_time as int) - CAST(signed_time as int)%180
)
SELECT
    dt, --日期
    site_agent_name,--代理区
    signed_site,--扫描网点
    CONCAT(deliver_user,'（',send_user_code,'）') AS deliver_user,--收派员
    SUM(total_cnt_per_three_minutes) AS total_deliver_cnt,--签收量
    SUM(IF(deliver_cnt_per_three_minute>=10,deliver_cnt_per_three_minute,0)) AS dense_deliver_cnt_per_three_minutes--3分钟以内集中签收票数
FROM res_pre
GROUP BY
    dt,
    site_agent_code,
    site_agent_name,
    signed_site_code,
    signed_site,
    send_user_code,
    deliver_user
;
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt